/**
 * Copyright (C) 2009+ Spark Zheng
 *
 * @file        test/test_tpool_enhance.c 
 * @date        2010-03-26
 * @brief	work like a pipeline
 *		CAUTION: not block the thread!
 *		so, dlib_buffer will be not recommended!
 * @see		dlib
 * @history
 */

#include <pthread.h>
#include <unistd.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>

#include "global.h"
#include "threadpool.h"
#include "buffer.h"

#define TP1_INIT_NUM	2
#define TP1_MAX_NUM	4

#define TP2_INIT_NUM	2
#define TP2_MAX_NUM	4

#define LOOP		1

void *tp1_task_cb(void *thread_data, void *arg);
void *tp2_task_cb(void *thread_data, void *arg);
void *tp1_timeout_cb(void *res_data, void *arg);
void *tp2_timeout_cb(void *res_data, void *arg);
void *tp1_result_cb(void *res_data, void *arg);
void *tp2_result_cb(void *res_data, void *arg);

struct test {
	char		*name;
	dlib_tpool_t	*tp1;
	dlib_tpool_t	*tp2;
};


int main(int argc, char **argv)
{
	UNUSED_PARAM(argc);
	UNUSED_PARAM(argv);

	int i;
	struct test t;
	t.name = "GLOBAL";
	t.tp1 = dlib_tpool_init(TP1_INIT_NUM, TP1_MAX_NUM, NULL, NULL, NULL);
	if (t.tp1 == NULL) {
		fprintf(stderr, "tpool1 init error.\n");
		return -1;
	}
	t.tp2 = dlib_tpool_init(TP2_INIT_NUM, TP2_MAX_NUM, NULL, NULL, NULL);
	if (t.tp2 == NULL) {
		fprintf(stderr, "tpool2 init error.\n");
		return -1;
	}

	dlib_buffer_t *buffer[LOOP];
	for (i = 0; i < LOOP; i++) {
		buffer[i] = dlib_buffer_init(16);
		buffer[i] = dlib_buffer_append(buffer[i], "INIT_TASK;", strlen("INIT_TASK;"));
	}

	/* set default callbacks */
	dlib_tpool_settask(t.tp1, tp1_task_cb, NULL);
	dlib_tpool_settask(t.tp2, tp2_task_cb, NULL);

	dlib_tpool_settotask(t.tp1, 1, tp1_timeout_cb, "TP1_TIMEOUT");
	dlib_tpool_setdetask(t.tp1, tp1_result_cb, &t);
	dlib_tpool_settotask(t.tp2, 20, tp2_timeout_cb, "TP2_TIMEOUT");
	dlib_tpool_setdetask(t.tp2, tp2_result_cb, &t);

	/* create listen threads for timeout result events */
#if 1
	pthread_t pid1, pid2, pid3, pid4;
	if (pthread_create(&pid1, NULL, (void *(*)(void *))dlib_tpool_timeout, (void *)t.tp1) < 0) {
		fprintf(stderr, "tpool1 create timeout callback error.\n");
		return -2;
	}
	if (pthread_create(&pid2, NULL, (void *(*)(void *))dlib_tpool_retrieve, (void *)t.tp1) < 0) {
		fprintf(stderr, "tpool1 create retrieve callback error.\n");
		return -2;
	}
	if (pthread_create(&pid3, NULL, (void *(*)(void *))dlib_tpool_timeout, (void *)t.tp2) < 0) {
		fprintf(stderr, "tpool2 create timeout callback error.\n");
		return -2;
	}
	if (pthread_create(&pid4, NULL, (void *(*)(void *))dlib_tpool_retrieve, (void *)t.tp2) < 0) {
		fprintf(stderr, "tpool2 create retrieve callback error.\n");
		return -2;
	}
	fprintf(stderr, "tpool callbacks[%lu, %lu, %lu, %lu].\n", pid1, pid2, pid3, pid4);
#else
	if (dlib_tpool_timeout_async(t.tp1) < 0) {
		fprintf(stderr, "tpool1 create timeout callback error.\n");
		return -2;
	}
	if (dlib_tpool_retrieve_async(t.tp1) < 0) {
		fprintf(stderr, "tpool1 create retrieve callback error.\n");
		return -2;
	}
	if (dlib_tpool_timeout_async(t.tp2) < 0) {
		fprintf(stderr, "tpool2 create timeout callback error.\n");
		return -2;
	}
	if (dlib_tpool_retrieve_async(t.tp2) < 0) {
		fprintf(stderr, "tpool2 create retrieve callback error.\n");
		return -2;
	}
#endif

	/* exec task */
	int ret = 0;
	for (i = 0; i < LOOP; i++) {
		ret = dlib_tpool_execute(t.tp1, 2, NULL, buffer[i]);
		fprintf(stderr, "tpool excute [%d] task return[%d].\n", i, ret);
	}

	dlib_tpool_wait(t.tp1);
	dlib_tpool_wait(t.tp2);

	dlib_tpool_dump(t.tp1, NULL);
	dlib_tpool_dump(t.tp2, NULL);
	
	dlib_tpool_exit(t.tp1);
	dlib_tpool_exit(t.tp2);

	for (i = 0; i < LOOP; i++) {
		dlib_buffer_exit(buffer[i]);
	}

	return 0;
}

void *tp1_task_cb(void *thread_data, void *arg)
{
	dlib_buffer_t *buffer = NULL;
	if (thread_data != NULL) {
		fprintf(stderr, "tp1[IN task_cb] thread data is %s.\n", (char *)thread_data);
	}
	if (arg != NULL) {
		buffer = (dlib_buffer_t *)arg;
		buffer = dlib_buffer_append(buffer, " IN_TASK1;", strlen(" IN_TASK1;"));
		fprintf(stderr, "tp1[IN task_cb] task data is %s.\n", buffer->buffer);
	}

	int num = random() % 50 + 100;

	usleep(num);
	
	fprintf(stderr, "tp1[IN task_cb] task exec time is %d.\n", num);

	return buffer;
}

void *tp2_task_cb(void *thread_data, void *arg)
{
	dlib_buffer_t *buffer = NULL;
	if (thread_data != NULL) {
		fprintf(stderr, "tp2[IN task_cb] thread data is %s.\n", (char *)thread_data);
	}
	if (arg != NULL) {
		buffer = (dlib_buffer_t *)arg;
		buffer = dlib_buffer_append(buffer, " IN_TASK2;", strlen(" IN_TASK2;"));
		fprintf(stderr, "tp2[IN task_cb] task data is %s.\n", buffer->buffer);
	}

	int num = random() % 20 + 200;

	usleep(num);
	
	fprintf(stderr, "tp2[IN task_cb] task exec time is %d.\n", num);

	return buffer;
}

void *tp1_timeout_cb(void *res_data, void *arg)
{
	if (res_data != NULL) {
		dlib_buffer_t *buffer = (dlib_buffer_t *)res_data;
		fprintf(stderr, "tp1[IN timeout_cb] result data is %s.\n", buffer->buffer);
	}
	if (arg != NULL) {
		fprintf(stderr, "tp1[IN timeout_cb] totask data is %s.\n", (char *)arg);
	}

	fprintf(stderr, "tp1[IN timeout_cb] this task is timeout, AND finished.\n");

	return NULL;
}

void *tp2_timeout_cb(void *res_data, void *arg)
{
	if (res_data != NULL) {
		dlib_buffer_t *buffer = (dlib_buffer_t *)res_data;
		fprintf(stderr, "tp2[IN timeout_cb] result data is %s.\n", buffer->buffer);
	}
	if (arg != NULL) {
		fprintf(stderr, "tp2[IN timeout_cb] totask data is %s.\n", (char *)arg);
	}

	fprintf(stderr, "tp2[IN timeout_cb] this task is timeout, AND finished.\n");

	return NULL;
}

void *tp1_result_cb(void *res_data, void *arg)
{
	dlib_buffer_t *buffer = NULL;
	if (res_data != NULL) {
		buffer = (dlib_buffer_t *)res_data;
		buffer = dlib_buffer_append(buffer, " IN_RESULT1;", strlen(" IN_RESULT1;"));
		fprintf(stderr, "tp1[IN result_cb] result data is %s.\n", buffer->buffer);
	}
	struct test *t;
	if (arg != NULL) {
		t = (struct test *)arg;
		fprintf(stderr, "tp1[IN result_cb] detask name is %s.\n", t->name);
	}

	int ret = dlib_tpool_execute(t->tp2, 2, NULL, (void *)buffer);
	fprintf(stderr, "tp1[IN result_cb] detask pass data to tp2[%d].\n", ret);

	return buffer;
}

void *tp2_result_cb(void *res_data, void *arg)
{
	dlib_buffer_t *buffer = NULL;
	if (res_data != NULL) {
		buffer = (dlib_buffer_t *)res_data;
		buffer = dlib_buffer_append(buffer, " IN_RESULT2;", strlen(" IN_RESULT2;"));
		fprintf(stderr, "tp2[IN result_cb] result data is %s.\n", buffer->buffer);
	}
	struct test *t;
	if (arg != NULL) {
		t = (struct test *)arg;
		fprintf(stderr, "tp2[IN result_cb] detask name is %s.\n", t->name);
	}

	fprintf(stderr, "tp2[IN result_cb] detask finished data.\n");

	return buffer;
}
